package com.guchenbo.example.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaConsumerTest2 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        consumer();
        //        multiConsumer();
    }

    private static void multiConsumer() {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 2; i++) {
            executorService.submit(() -> {
                consumer();
            });
        }
    }

    private static void consumer() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        properties.put("group.id", "test_log4");
//        properties.put("enable.auto.commit", "false");
        properties.put("max.poll.records", "1000");
        properties.put("auto.offset.reset", "earliest");

        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        ProducerConfig producerConfig;
        KafkaConsumer kafkaConsumer = new KafkaConsumer(properties);
        kafkaConsumer.subscribe(Arrays.asList("part4"));

        //        ConsumerRecords<String, String> records2 = kafkaConsumer.poll(100);
        //        System.out.println(records2);
        //        System.out.println(records2.count());
        //        for (ConsumerRecord<String, String> record : records2) {
        //            System.out.println(record);
        //        }
        int i = 0;
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(10).toMillis());
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(i++);
                String msg = record.value();
                //                if (msg.contains("test log")) {
                System.out.println(msg);
                //                }
            }
            // 模拟慢操作
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
//            kafkaConsumer.commitSync();
        }
    }

}
